#include "shmipc/com/internal/uds_session.h"

#include "shmipc/core/xtime.h"
#include "shmipc/com/internal/uds_com.h"
#include "shmipc/com/internal/xsocket.h"

#define GET_SEND_MSG_TIMEOUT_MS (10)
#define SOCK_RECV_TIMEOUT_MS (10)


namespace shmipc::com::internal {
using namespace shmipc::core;

static void DefaultOnRecvHandler(XMessageSp msg) {
    static_cast<void>(msg);
    //do nothing
}

static void DefaultOnErrHandler(XErrC err) {
    static_cast<void>(err);
    //do nothing
}

UdsSession::~UdsSession(){
    stop();
    if (on_release_handler_) {
        on_release_handler_(shared_from_this());
    }
}

UdsSession::UdsSession(XComId session_id, const std::string& session_name, OnSessionReleaseHandler on_release /*= nullptr */ )
: session_id_(session_id), session_name_(session_name), on_release_handler_(on_release) {
    /** todo something */
}

XResultCode UdsSession::start() {
    if ((session_id_ == XBUS_INVALID_XCOM_ID) || send_worker_ || recv_worker_) {
        return XResultCode::FromError(XErrC::InvalidStatus);
    }

    send_queue_ = XMessageQueueCreate();
    send_worker_ = std::make_unique<XThread>("OnSend:" + session_name_, [&](){ this->onSendWorker(); });
    recv_worker_ = std::make_unique<XThread>("OnRecv:" + session_name_, [&](){ this->onRecvWorker(); });

    if (!send_queue_ || !send_worker_ || !recv_worker_) {
        send_queue_ = nullptr;
        send_worker_.reset(nullptr);
        recv_worker_.reset(nullptr);
        return XResultCode::FromError(XErrC::InsufficientReousces);
    }

    is_running_ = true;
    if ((send_queue_->start() != XErrC::OK) || !send_worker_->start() || !recv_worker_->start()) {
        is_running_ = false;
        send_worker_->join();
        recv_worker_->join();
        send_queue_ = nullptr;
        send_worker_.reset(nullptr);
        recv_worker_.reset(nullptr);
        return XResultCode::FromError(XErrC::Undefined);    
    }

    return { };
}

XResultCode UdsSession::stop() {
    if (!send_worker_ || !recv_worker_) {
        return XResultCode::FromError(XErrC::InvalidStatus);
    }

    is_running_ = false;
    send_worker_->join();
    recv_worker_->join();
    send_queue_ = nullptr;
    send_worker_.reset(nullptr);
    recv_worker_.reset(nullptr);
    session_id_ = XBUS_INVALID_XCOM_ID;
    return { };
}

XResultCode UdsSession::reset(XComId session_id){ /**  first clean send and receive queue, then update session id */
    if (!send_worker_ || !recv_worker_) {
        return XResultCode::FromError(XErrC::InvalidStatus);
    }

    send_queue_->stop();
    send_queue_->clear();
    session_id_ = session_id;
    send_queue_->start();

    return { };
}

void UdsSession::setOnRecvHandler(OnReceiveHandler on_handler) {
    on_recv_handler_ = on_handler ? on_handler : DefaultOnRecvHandler;
}

void UdsSession::setOnErrorHandler(OnErrorHandler on_handler) {
    on_err_handler_ = on_handler ? on_handler : DefaultOnErrHandler;
}

XResultCode UdsSession::send(XMessageSp msg) {
    if (!send_queue_) {
        return XResultCode::FromError(XErrC::InvalidStatus);
    }

    return send_queue_->push(msg);
}

void UdsSession::onSendWorker() {
    //todo, refine me, to suspend current thread when specification error(e.g XErrC::Disconnected) occur.
    while (is_running_) {
        auto msg_res = send_queue_->pop(GET_SEND_MSG_TIMEOUT_MS); 
        if (msg_res) {
            auto msg = msg_res.value();
            auto res = XSocketSend(session_id_, msg->data(), msg->length());

            if (!res) {
                //todo process error.
                on_err_handler_(res.error());
            }
        } else {
            auto errc = msg_res.error();
            switch (errc) {
            case XErrC::InvalidStatus: 
                SleepMs(GET_SEND_MSG_TIMEOUT_MS);
                break;
            case XErrC::Disconnected:
                //todo process error
                break;
            default:
                break;
            }
        }
    }
}

void UdsSession::onRecvWorker() {
    //todo, refine me, to suspend current thread when specification error(e.g XErrC::Disconnected) occur.
    while (is_running_) { 
        auto res = UdsRecvMsg(session_id_, SOCK_RECV_TIMEOUT_MS); 
        if (res) {
            on_recv_handler_(res.value());
        } else {
            auto errc = res.error();
            on_err_handler_(errc);
            core::SleepMs(SOCK_RECV_TIMEOUT_MS);
        }
    }
}

} //namespace shmipc::com::internal